package localmq

import (
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	prototypes "hundsun.com/hsl/hschain/protos/types"
	"hundsun.com/hsl/hschain/store/mq"

	"unsafe"

	"hundsun.com/hsl/hschain/common/types"
)

// 消息队列的主要作用是解耦合，让各个模块相对的独立运行。
// 每个模块可有多个client对象
// 主要的操作如下：
// Client := queue.Client("")
// Client.Sub("topicName")
// for msg := range Client.Recv() {
//     process(msg)
// }
// process 函数会调用 处理具体的消息逻辑

const (
	topicSeparator = "."
	idSeparator    = "-"
)

var gid int64 // 用作msgId, 全局单调递增
var tid int64 // 用作topic模式的topicId, 全局单调递增

// Client 消息队列的接口，每个模块都需要一个发送接收client
type Client interface {
	Send(msg *Message, waitReply bool) (err error) // 发送消息
	SendTimeout(msg *Message, waitReply bool, timeout time.Duration) (err error)
	Wait(msg *Message) (*Message, error)                               // 等待消息处理完成 获取消息回复
	WaitTimeout(msg *Message, timeout time.Duration) (*Message, error) // 等待消息处理完成 获取消息回复
	Recv() chan *Message
	Reply(msg *Message)
	Sub(mainTopic, subTopic string) error // 订阅消息
	Close()
	CloseQueue() (*prototypes.Reply, error)
	NewMessage(topic string, ty int64, data interface{}) (msg *Message)
	FreeMessage(msg ...*Message) // 回收msg 需要注意上下文不再引用
}

type client struct {
	q         *queue
	recv      chan *Message
	done      chan struct{}
	wg        *sync.WaitGroup
	topic     unsafe.Pointer
	isClosed  int32
	isClosing int32
	mode      string // 订阅模式
}

// newClient 新建客户端实例
func newClient(q *queue, mode string) Client {
	client := &client{}
	client.q = q
	client.recv = make(chan *Message, 5)
	client.done = make(chan struct{}, 1)
	client.wg = &sync.WaitGroup{}
	client.mode = mode
	return client
}

// Send 发送消息, msg 消息, waitReply 是否等待回应(worker模式)
// 1. 系统保证send出去的消息就是成功了，除非系统崩溃
// 2. (worker模式)系统保证每个消息都有对应的 response 消息
func (client *client) Send(msg *Message, waitReply bool) (err error) {
	timeout := time.Duration(-1)
	err = client.SendTimeout(msg, waitReply, timeout)
	if err == ErrQueueTimeout {
		panic(err)
	}
	return err
}

// SendTimeout 超时发送, msg 消息, waitReply 是否等待回应, timeout 超时时间
func (client *client) SendTimeout(msg *Message, waitReply bool, timeout time.Duration) (err error) {
	if client.isClose() {
		return ErrIsQueueClosed
	}

	switch client.mode {
	case mq.ModeWorker.String():
		if waitReply {
			return client.q.send(msg, timeout)
		}
		//msg.chReply = nil
		return client.q.sendLowTimeout(msg, timeout)
	case mq.ModeTopic.String():
		// topic模式：只支持低优先级发送（异步），不支持同步发送、消息回复
		return client.q.sendLowTimeoutWithTopicMode(msg, timeout)
	default:
		return types.ErrNotFoundMode
	}
}

// 两种优先级别的消息发送 1. SendAsync 低优先级(异步) 2. Send 高优先级别的发送消息(同步)

// NewMessage 新建消息 topic模块名称 ty消息类型 data 数据
func (client *client) NewMessage(topic string, ty int64, data interface{}) (msg *Message) {
	id := atomic.AddInt64(&gid, 1)
	msg = client.q.msgPool.Get().(*Message)
	msg.ID = id
	msg.Ty = ty
	msg.Data = data
	msg.Topic = topic
	return
}

// FreeMessage 回收msg 到内存池
func (client *client) FreeMessage(msgs ...*Message) {
	for _, msg := range msgs {
		if msg == nil || msg.chReply == nil {
			continue
		}
		msg.Data = nil
		client.q.msgPool.Put(msg)
	}
}

// Reply 消息应答
func (client *client) Reply(msg *Message) {
	if msg.chReply != nil {
		msg.Reply(msg)
		return
	}
	if msg.callback != nil {
		client.q.callback <- msg
	}
}

// WaitTimeout 等待时间 msg 消息 timeout 超时时间
func (client *client) WaitTimeout(msg *Message, timeout time.Duration) (*Message, error) {
	if msg.chReply == nil && msg.chMsgReply == nil {
		return &Message{}, errors.New("empty wait channel")
	}

	var t <-chan time.Time
	if timeout > 0 {
		timer := time.NewTimer(timeout)
		defer timer.Stop()
		t = timer.C
	}

	// receive block when timeout <= 0
	select {
	case m := <-msg.chMsgReply:
		topic, _ := genTopicKey(m.MainTopic, m.SubTopic)
		c := &Message{Topic: topic, Ty: m.Ty, Data: m.Data, chMsgReply: m.ChReply}
		return c, c.Err()
	case m := <-msg.chReply:
		return m, m.Err()
	case <-client.done:
		return &Message{}, ErrIsQueueClosed
	case <-t:
		return &Message{}, ErrQueueTimeout
	}
}

// Wait 等待时间
func (client *client) Wait(msg *Message) (*Message, error) {
	timeout := time.Duration(-1)
	m, err := client.WaitTimeout(msg, timeout)
	if err == ErrQueueTimeout {
		panic(err)
	}
	return m, err
}

// Recv 获取接受消息通道
func (client *client) Recv() chan *Message {
	return client.recv
}

func (client *client) getTopic() string {
	return *(*string)(atomic.LoadPointer(&client.topic))
}

func (client *client) setTopic(topic string) {
	// #nosec
	atomic.StorePointer(&client.topic, unsafe.Pointer(&topic))
}

func (client *client) isClose() bool {
	return atomic.LoadInt32(&client.isClosed) == 1
}

func (client *client) isInClose() bool {
	return atomic.LoadInt32(&client.isClosing) == 1
}

// Close 关闭client
func (client *client) Close() {
	if atomic.LoadInt32(&client.isClosed) == 1 || atomic.LoadPointer(&client.topic) == nil {
		return
	}
	topic := client.getTopic()
	client.q.closeTopic(topic)
	//client.q.closeTopicSub(topic)
	//client.q.closeWorkerSub(topic)
	close(client.done)
	atomic.StoreInt32(&client.isClosing, 1)
	client.wg.Wait()
	atomic.StoreInt32(&client.isClosed, 1)
	close(client.Recv())
	for msg := range client.Recv() {
		msg.Reply(client.NewMessage(msg.Topic, msg.Ty, types.ErrChannelClosed))
	}
}

// CloseQueue 关闭消息队列
func (client *client) CloseQueue() (*prototypes.Reply, error) {
	//	Client.q.Close()
	if client.q.isClosed() {
		return &prototypes.Reply{IsOk: true}, nil
	}
	// log.Debug("queue", "msg", "closing")
	client.q.interrupt <- struct{}{}
	//	close(Client.q.interrupt)
	return &prototypes.Reply{IsOk: true}, nil
}

func (client *client) isEnd(data *Message, ok bool) bool {
	if !ok {
		return true
	}
	if data.Data == nil && data.ID == 0 && data.Ty == 0 {
		return true
	}
	if atomic.LoadInt32(&client.isClosed) == 1 {
		return true
	}
	return false
}

// Sub 订阅消息
func (client *client) Sub(mainTopic, subTopic string) error {
	//正在关闭或者已经关闭
	if client.isInClose() || client.isClose() {
		return nil
	}
	client.wg.Add(1)

	var topic string
	switch client.mode {
	case mq.ModeWorker.String():
		topic, _ = genTopicKey(mainTopic, subTopic)
	case mq.ModeTopic.String():
		topic, _ = genUniqueTopicKey(mainTopic, subTopic)
	default:
		topic, _ = genTopicKey(mainTopic, subTopic)
	}

	client.setTopic(topic)
	sub := client.q.chanSub(topic)
	go func() {
		defer func() {
			client.wg.Done()
		}()
		for {
			select {
			case data, ok := <-sub.high:
				if client.isEnd(data, ok) {
					// log.Info("unsub1", "topic", topic)
					return
				}
				client.Recv() <- data
			default:
				select {
				case data, ok := <-sub.high:
					if client.isEnd(data, ok) {
						// log.Info("unsub2", "topic", topic)
						return
					}
					client.Recv() <- data
				case data, ok := <-sub.low:
					if client.isEnd(data, ok) {
						// log.Info("unsub3", "topic", topic)
						return
					}
					client.Recv() <- data
				case <-client.done:
					// log.Error("unsub4", "topic", topic)
					return
				}
			}
		}
	}()

	return nil
}

func genTopicKey(mainTopic, subTopic string) (string, error) {
	if strings.Contains(mainTopic, topicSeparator) {
		return "", types.ErrTopicContainReservedChar
	}
	if strings.Contains(subTopic, topicSeparator) {
		return "", types.ErrTopicContainReservedChar
	}

	topic := mainTopic
	if len(subTopic) > 0 {
		topic = mainTopic + topicSeparator + subTopic
	}
	return topic, nil
}

func genUniqueTopicKey(mainTopic, subTopic string) (string, error) {
	topic, err := genTopicKey(mainTopic, subTopic)
	if err != nil {
		return "", err
	}

	id := atomic.AddInt64(&tid, 1)
	return fmt.Sprintf("%s%s%d", topic, idSeparator, id), nil
}

func splitTopicKey(topic string) (mainTopic, subTopic string) {
	msgs := strings.SplitN(topic, topicSeparator, 2)
	if len(msgs) > 1 {
		return msgs[0], msgs[1]
	}
	return msgs[0], ""
}
